package com.desheng.bigdata.flink.table

import com.alibaba.fastjson.JSON
import com.desheng.bigdata.flink.domain.User
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, _}
import org.apache.flink.table.api.scala.BatchTableEnvironment
import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.types.Row
/**
  * Flink table的udf操作
  * 用户浏览的数据
  * {"userID": 2, "eventTime": "2020-10-01 10:02:00", "eventType": "browse", "productID": "product_5", "productPrice": 20.99}
  * 按照userID,来对用户的浏览的productPrice进行求和统计
  */
object _08FlinkTable2UDAFOps {
    def main(args: Array[String]): Unit = {
        //创建批对应的Env
        val batchEnv = ExecutionEnvironment.getExecutionEnvironment
        //基于批Env构建BatchTableEnv
        val tblEnv = BatchTableEnvironment.create(batchEnv)

        val userDataSet: DataSet[User] = batchEnv.readTextFile("file:/E:/data/flink/flink-udf.json")
                .map(line => JSON.parseObject(line, classOf[User]))

        val userTbl = tblEnv.fromDataSet(userDataSet)
        tblEnv.registerTable("userTbl", userTbl)
        //注册用户自定义函数
        tblEnv.registerFunction("mySum", new MySumUDAF())

        val table = tblEnv.sqlQuery(
                """
                  |select
                  |  userID,
                  |  sum(productPrice) sum1,
                  |  mySum(productPrice) sum2
                  |from userTbl
                  |group by userID
                """.stripMargin)

        tblEnv.toDataSet[Row](table).print()
    }
}

/**
  * 自定义udaf来实现sum的功能
  * [T, ACC]
  * 其中T就是返回值的类型
  * ACC就是该udaf在计算过程中临时变量的类型，通过累加器的方式完成封装
  */
class MySumUDAF extends AggregateFunction[Double, MySumAcc] {
    /**
      * 获取最后的udaf的结果值
      * @param accumulator 在udaf中用来存储中间结果的累加器
      */
    override def getValue(accumulator: MySumAcc): Double = {
        accumulator.sumPrice
    }

    /**
      * 完成udaf的累加的过程
      * @param accumulator 不断更新该累加器中的值
      * @param productPrice 被累加的值
      */
    def accumulate(accumulator: MySumAcc, productPrice: Double): Unit ={
        accumulator.sumPrice += productPrice
    }

    /**
      * 初始化用于进行累加的累加器
      */
    override def createAccumulator(): MySumAcc = {
        val mySumAcc = new MySumAcc
        mySumAcc.sumPrice = 0
        mySumAcc
    }

    /**
      * 重置累加器
      */
    def resetAccumulator(accumulator: MySumAcc):Unit = {
        accumulator.sumPrice = 0
    }
}
/*
    自定义一个累加器，用于在udaf中临时存储累加的结果
 */
class MySumAcc extends Serializable {
    var sumPrice: Double = 0
}